package org.fastsyncer.worker.framework;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.parser.framework.ParserFactory;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Java 线程池 Java通过Executors提供四种线程池，分别为：
 * newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
 * newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。 newScheduledThreadPool
 * 创建一个定长线程池，支持定时及周期性任务执行。 newSingleThreadExecutor
 * 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
 * 
 * @author YangPei
 * @date 2017年5月13日 下午7:16:37
 * @version 1.0.0
 */
public class WorkerThreadPool {

	private static Logger logger = LoggerFactory.getLogger(WorkerThreadPool.class);

	/**
	 * CountDownLatch，一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。
	 */
	private CountDownLatch latch;
	
	/**
	 * 存储所有Task,每个Task里面包含执行的任务
	 */
	private ConcurrentLinkedQueue<Integer[]> workerQuene;

	/**
	 * 记录执行成功数
	 */
	private AtomicInteger excute = new AtomicInteger(0);
	
	/**
	 * 记录执行失败数
	 */
	private AtomicInteger success = new AtomicInteger(0);

	/**
	 * 失败原因
	 */
	private ConcurrentLinkedQueue<JSONObject> reasonQuene = new ConcurrentLinkedQueue<>();

	/**
	 * 处理同步任务
	 * 
	 * @param mappingTask
	 *            同步任务
	 * @param tasks
	 *            任务清单,[{0,999},{1000,1999},...]
	 * @throws InterruptedException
	 * @throws JSONException
	 */
	public JSONObject sync(MappingTask mappingTask, ConcurrentLinkedQueue<Integer[]> tasks)
			throws InterruptedException, JSONException {
		// 获取要执行的任务
		workerQuene = tasks;

		// 获取配置的线程个数
		Integer threadNum = mappingTask.getThreadNum();
		// 当前任务个数
		int tSize = tasks.size();
		// 为了节省不必要的线程资源，当线程个数超过实际处理的任务，那么只需要开辟处理的任务个数即可
		threadNum = threadNum > tSize ? tSize : threadNum;
		
		// 线程个数
		latch = new CountDownLatch(threadNum);

		// 创建线程池
		ExecutorService es = Executors.newCachedThreadPool();
		// 分配配置中设置的线程数
		for (int i = 0; i < threadNum; i++) {
			es.execute(new Runnable() {

				@Override
				public void run() {
					while (!workerQuene.isEmpty()) {
						// 消费
						Integer[] pos = workerQuene.poll();
						if(pos==null || pos.length != 2){
							break;
						}
						try {
							Integer[] result = ParserFactory.getInstance().parse(mappingTask, pos, reasonQuene);
							// 累加执行总数、成功条数
							excute.getAndAdd(result[0]);
							success.getAndAdd(result[1]);
						} catch (Exception e) {
			                try {
			                    // 记录错误日志
			                    JSONObject err = new JSONObject();
                                err.put("cause", e.getClass());
                                err.put("message", e.getMessage());
                                reasonQuene.offer(err);
                            } catch (JSONException e1) {
                            }
						}
					}
					latch.countDown();
				}

			});
		}
		// 使得主线程(main)阻塞直到latch.countDown()为零才继续执行
		latch.await();
		es.shutdown();

		// 统计错误原因，因为考虑产生的错误不会很多，size不会影响性能
		JSONArray reasonArr = new JSONArray();
		int i = 0;
		while (!reasonQuene.isEmpty()) {
			JSONObject err = reasonQuene.poll();
			reasonArr.put(i, err);
			logger.info("Record error(" + (i + 1) + "):" + err);
			i++;
		}

		// 统计执行结果
		JSONObject result = new JSONObject();
		result.put("excute", excute.get());
		result.put("success", success.get());
		result.put("reason", reasonArr);
		return result;
	}

}
